package com.hmall.loadbalancer.Impl;

import com.hmall.loadbalancer.AbstractLoadBalance;
import com.hmall.remoting.dto.RpcRequest;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;

public class ConsistentHashLoadBalance extends AbstractLoadBalance {

    private final ConcurrentHashMap<String, ConsistentHashSelector> selectors = new ConcurrentHashMap<>();//服务名和选择器的映射

    @Override
    protected String doSelect(List<String> serviceAddresses, RpcRequest rpcRequest) {
        String rpcServiceName = rpcRequest.getRpcServiceName();
        int identityHashCode = System.identityHashCode(serviceAddresses);//获取服务列表的哈希值
        ConsistentHashSelector selector = selectors.get(rpcServiceName);//获取选择器
        if (selector == null || identityHashCode != selector.identityHashCode) {//判断服务列表是否有修改，若有修改则选择器的哈希值与服务列表的哈希值不一致（相同的服务列表哈希值应该相同）
            selectors.put(rpcServiceName,new ConsistentHashSelector(serviceAddresses,160,identityHashCode));//重构映射
            selector = selectors.get(rpcServiceName);
        }
        return selector.select(rpcServiceName + Arrays.stream(rpcRequest.getParameters()));
    }

    static class ConsistentHashSelector {//每个服务的负载均衡选择器
        private final TreeMap<Long, String> virtualInvokers;//哈希值与服务地址的映射

        private final int identityHashCode;

        ConsistentHashSelector(List<String> invokers, int replicaNumber, int identityHashCode) {
            this.virtualInvokers = new TreeMap<>();
            this.identityHashCode = identityHashCode;

            for (String invoker : invokers) {//遍历服务列表
                for (int i = 0; i < replicaNumber/4; i++) {
                    byte[] digest = md5(invoker);
                    for (int j = 0; j < 4; j++) {
                        long hash = hash(digest, j);//将一个digest分为4个，减少MD5运算
                        virtualInvokers.put(hash, invoker);//每个服务地址虚拟出160个虚拟节点
                    }
                }
            }
        }

        static byte[] md5(String address) {
            MessageDigest md;
            try {
                md = MessageDigest.getInstance("MD5");
                md.update(address.getBytes());
            } catch (NoSuchAlgorithmException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            return md.digest();
        }

        static long hash(byte[] bytes,int idx) {
            /*  & 255：确保字节值为无符号（Java 中 byte 是有符号的）。
                << 24/16/8：将 4 字节组合成一个 32 位整数。
                & 4294967295L：将结果转为无符号（Java 中 long 是有符号的）
             */
            return (long) bytes[idx * 4 + 3] & 255L << 24 | (long) bytes[idx * 4 + 2] & 255 << 16 | (long) bytes[idx * 4 + 1] & 255 << 8 | (long) bytes[idx * 4] & 255 & 4294967295L;
        }

        public String select(String rpcServiceKey){
            byte[] digest = md5(rpcServiceKey);
            return selectForKey(hash(digest, 0));
        }

        public String selectForKey(long hash) {//根据服务名称和参数的哈希值来获取指定地址
            Map.Entry<Long, String> entry = virtualInvokers.tailMap(hash, true).firstEntry();
            if (entry == null) {
                entry = virtualInvokers.firstEntry();
            }
            return entry.getValue();
        }
    }
}
